package com.study.flink.java.day06_exactly;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.RandomAccessFile;

/**
 * 自定义并行的source
 */
public class MyParFileSource extends RichParallelSourceFunction<Tuple2<String, String>> {

    private String path;
    private Boolean flag = true;

    public MyParFileSource() {
    }

    public MyParFileSource(String path) {
        this.path = path;
    }

    // 就执行一次
    @Override
    public void run(SourceContext<Tuple2<String, String>> sourceContext) throws Exception {

        int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
        // var/data/0.txt
        RandomAccessFile randomAccessFile = new RandomAccessFile(path + "/" + subtaskIndex + ".txt", "r");
        while (flag) {
            String line = randomAccessFile.readLine();
            if (line != null) {
                line = new String(line.getBytes("ISO-8859-1"), "UTF-8");
                // 将数据发送出去
                sourceContext.collect(Tuple2.of(subtaskIndex + "", line));
            } else {
                Thread.sleep(1000L);
            }
        }

    }

    @Override
    public void cancel() {
        flag = false;
    }


}
